package com.ylkj.bus.device.report;

import com.hwl.redis.RedisConfig;

import com.ylkj.bus.monitor.equipment.model.EquipmentAlarmMonitorLogEnty;

import com.ylkj.dao.RdsUtils;
import com.ylkj.io.stream.rds.RdsConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.BeanListHandler;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;


import java.sql.*;
import java.text.MessageFormat;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @Author: wengzx
 * @Date: 2025/4/23
 * @Desc:
 */
@Slf4j
public class UpdateDevicMonitorAndUnpack extends ProcessFunction<EquipmentAlarmMonitorLogEnty,String> {
    // SQL模板常量
    private static final String DEVICE_INIT_SQL = "SELECT device_code as deviceCode, product_id as productId FROM {0}  LIMIT ? OFFSET ?";
    private static final String PRODUCT_INIT_SQL = "SELECT product_id as productId, product_type as productType FROM {0}  LIMIT ? OFFSET ?";
    private static final String INSERT_MONITOR_SQL = "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`, `event_value`, `alarm_time`, `blanking_tool`, `product_id`,`product_type`,`message_suggestion`,`alert_source`,`net_code`,`personnel_handler`,`handling_time`,`alert_description`,create_time, update_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now(), now()) ON DUPLICATE KEY UPDATE `alarm_time` = VALUES(`alarm_time`), `update_time` = now()";
    private static final String UPDATE_ONLINE_SQL = "UPDATE {0} SET `alarm_cover` = ?, `message_suggestion` = ?, `personnel_handler` = ?, `handling_time` = ?, `update_time` = now() WHERE device_code = ? AND event_type = ?";
    private static final String UPDATE_LOG_SQL = "UPDATE {0} SET `alarm_count` = ? WHERE device_code = ? AND event_type = ? AND alarm_cover = 1";
    private static final String INSERT_LOG_SQL = "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`, `event_value`, `alarm_time`, `blanking_tool`, `product_id`,`product_type`,`message_suggestion`,`alert_source`,`net_code`,`personnel_handler`,`handling_time`,`alert_description`,`alarm_count`, create_time, update_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now(), now())";
    private static final String CHECK_LOG_EXISTS_SQL = "SELECT 1 FROM {0} WHERE device_code = ? AND event_type = ? AND alarm_cover = 1";

    // 配置常量
    private static final int MAX_RETRIES = 3;                // 最大重试次数
    private static final int PAGE_SIZE = 3000;               // 优化后的分页大小
    private static final int DEADLOCK_ERROR_CODE = 1213;     // MySQL死锁错误码

    // 依赖注入
    private final RdsConfig rdsConfig;
    private final RedisConfig redisConfig;
    private final String monitorTableName;     // 监控主表
    private final String logTableName;         // 日志表
    private final String deviceMetaTable;      // 设备元数据表
    private final String productMetaTable;     // 产品元数据表

    // 缓存容器（优化初始容量）
    private final Map<String, Long> deviceCache = new HashMap<>(16384);
    private final Map<Long, String> productCache = new HashMap<>(1024);

    // 数据库工具
    private RdsUtils rdsUtils;
    private Jedis jedis;

    // 预编译SQL语句
    private String insertMonitorSql;
    private String updateOnlineSql;
    private String updateLogSql;
    private String insertLogSql;
    private String checkLogExistsSql;

    /**
     * 构造函数，初始化表名和配置
     * @param monitorTableName 监控主表名称
     * @param logTableName 日志表名称
     * @param deviceMetaTable 设备元数据表名称
     * @param productMetaTable 产品元数据表名称
     */
    public UpdateDevicMonitorAndUnpack(String monitorTableName, RdsConfig rdsConfig,
                                    RedisConfig redisConfig ,String logTableName,
                                    String deviceMetaTable, String productMetaTable)
    {
        this.monitorTableName = monitorTableName;
        this.logTableName = logTableName;
        this.deviceMetaTable = deviceMetaTable;
        this.productMetaTable = productMetaTable;
        this.rdsConfig = rdsConfig;
        this.redisConfig = redisConfig;
    }

    /**
     * 初始化方法：建立数据库连接，预编译SQL，加载缓存
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        // 初始化数据库工具
        rdsUtils = new RdsUtils(rdsConfig);
        jedis = redisConfig.getJedis();

        // 预编译SQL语句
        insertMonitorSql = MessageFormat.format(INSERT_MONITOR_SQL, monitorTableName);
        updateOnlineSql = MessageFormat.format(UPDATE_ONLINE_SQL, monitorTableName);
        updateLogSql = MessageFormat.format(UPDATE_LOG_SQL, logTableName);
        insertLogSql = MessageFormat.format(INSERT_LOG_SQL, logTableName);
        checkLogExistsSql = MessageFormat.format(CHECK_LOG_EXISTS_SQL, logTableName);

        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(
                this::refreshCaches,   // 要执行的任务
                0,                     // 初始延迟（立即执行）
                1,                     // 周期长度
                TimeUnit.DAYS          // 时间单位（天）
        ); // 每天刷新1次刷新

        initDeviceCache();
        initProductCache();

    }

    private void refreshCaches() {
        initDeviceCache();
        initProductCache();
        log.info("缓存已刷新，设备缓存大小: {}, 产品缓存大小: {}", deviceCache.size(), productCache.size());
    }

    /**
     * 核心处理方法，处理设备上下线事件
     */
    @Override
    public void processElement(EquipmentAlarmMonitorLogEnty entry,
                               Context ctx, Collector<String> out) throws Exception {
        String deviceCode = entry.getDeviceCode();
        String eventType = entry.getEventType();

        // 元数据校验
        Long productId = deviceCache.get(deviceCode);
        String productType = productCache.get(productId);
        if (productId == null || StringUtils.isBlank(productType)) {
            log.warn("设备元数据缺失，跳过处理 deviceCode: {}", deviceCode);
            return;
        }

        // 事件类型标准化处理
        if ("deviceOffline".equalsIgnoreCase(eventType)) {
            handleOfflineEvent(entry, deviceCode, productId, productType);
        } else if ("deviceOnline".equalsIgnoreCase(eventType)) {
            handleOnlineEvent(entry, deviceCode);
        } else {
            log.warn("未知事件类型: {}", eventType);
        }
    }

    /**
     * 处理设备离线事件（带死锁重试）
     */
    private void handleOfflineEvent(EquipmentAlarmMonitorLogEnty entry,
                                    String deviceCode, Long productId, String productType) {
        int retryCount = 0;
        while (retryCount < MAX_RETRIES) {
            try (Connection conn = rdsUtils.getConnection();
                 AutoCloseable ignored = () -> conn.rollback()) { // 自动回滚包装

                // 开启事务
                conn.setAutoCommit(false);

                // 插入监控记录
                boolean inserted = insertMonitorData(conn, entry, deviceCode, productId, productType);
                if (inserted) {
                    // 检查日志是否存在
                    boolean exists = checkLogExists(conn, deviceCode, "connection_abnormal");
                    if (exists) {
                        updateMonitorLog(conn, entry, deviceCode);
                    } else {
                        insertMonitorLog(conn, entry, deviceCode, productId, productType);
                    }
                }

                conn.commit();
                break;

            } catch (Exception e) {
                handleSQLException((SQLException) e, retryCount, deviceCode);
                retryCount++;
            }
        }
    }

    /**
     * 处理设备上线事件（带死锁重试）
     */
    private void handleOnlineEvent(EquipmentAlarmMonitorLogEnty entry, String deviceCode) {
        int retryCount = 0;
        while (retryCount < MAX_RETRIES) {
            try (Connection conn = rdsUtils.getConnection();
                 AutoCloseable ignored = () -> conn.rollback()) {

                conn.setAutoCommit(false);

                // 更新在线状态
                boolean updated = updateOnlineStatus(conn, entry, deviceCode);
                if (updated) {
                    // 清除Redis中的离线标记
                    jedis.hdel(redisConfig.getTargetKey(), deviceCode);
                    // 更新日志记录
                    updateOnlineLog(conn, entry, deviceCode);
                }

                conn.commit();
                break;

            } catch (Exception e) {
                handleSQLException((SQLException) e, retryCount, deviceCode);
                retryCount++;
            }
        }
    }

    /**
     * 初始化设备缓存（优化分页策略）
     */
    private void initDeviceCache() {
        try (Connection conn = rdsUtils.getConnection()) {
            String sql = MessageFormat.format(DEVICE_INIT_SQL, deviceMetaTable);
            QueryRunner runner = new QueryRunner();

            int page = 1;
            while (true) {
                int offset = (page - 1) * PAGE_SIZE;
                List<DeviceMoDO> devices = runner.query(
                        conn, sql, new BeanListHandler<>(DeviceMoDO.class),
                        PAGE_SIZE, offset
                );

                if (devices.isEmpty()) break;
                devices.forEach(d -> deviceCache.put(d.getDeviceCode(), d.getProductId()));
                page++;
            }

            log.info("设备缓存加载完成，总记录数: {}", deviceCache.size());
        } catch (Exception e) {
            log.error("设备缓存初始化失败", e);
        }
    }

    /**
     * 初始化产品缓存
     */
    private void initProductCache() {
        try (Connection conn = rdsUtils.getConnection()) {
            String sql = MessageFormat.format(PRODUCT_INIT_SQL, productMetaTable);
            QueryRunner runner = new QueryRunner();

            int page = 1;
            while (true) {
                int offset = (page - 1) * PAGE_SIZE;
                List<ProductDO> products = runner.query(
                        conn, sql, new BeanListHandler<>(ProductDO.class),
                        PAGE_SIZE, offset
                );

                if (products.isEmpty()) break;
                products.forEach(p -> productCache.put(p.getProductId(), p.getProductType()));
                page++;
            }

            log.info("产品缓存加载完成，总记录数: {}", productCache.size());
        } catch (Exception e) {
            log.error("产品缓存初始化失败", e);
        }
    }

    /**
     * 插入监控主表数据
     */
    private boolean insertMonitorData(Connection conn,
                                      EquipmentAlarmMonitorLogEnty entry,
                                      String deviceCode, Long productId,
                                      String productType) throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(insertMonitorSql)) {
            // 参数绑定（索引优化）
            stmt.setString(1, deviceCode);
            stmt.setString(2, entry.getAlertLevel());
            stmt.setString(3, entry.getAlertDimensions());
            stmt.setString(4, entry.getAlarmCover());
            stmt.setString(5, "connection_abnormal");
            stmt.setString(6, entry.getEventValue());
            stmt.setTimestamp(7, convertToSqlDate(entry.getAlarmTime()));
            stmt.setString(8, entry.getBlankingTool());
            stmt.setLong(9, productId);
            stmt.setString(10, productType);
            stmt.setString(11, entry.getMessageSuggestion());
            stmt.setString(12, entry.getAlertSource());
            stmt.setString(13, entry.getNetCode());
            stmt.setString(14, entry.getPersonnelHandler());
            stmt.setTimestamp(15, convertToSqlDate(entry.getHandlingTime()));
            stmt.setString(16, entry.getAlertDescription());

            return stmt.executeUpdate() > 0;
        } catch (SQLException e) {
            logSqlError(e, "插入监控数据失败", deviceCode);
            return false;
        }
    }

    /**
     * 检查日志是否存在
     */
    private boolean checkLogExists(Connection conn, String deviceCode, String eventType)
            throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(checkLogExistsSql)) {
            stmt.setString(1, deviceCode);
            stmt.setString(2, eventType);
            try (ResultSet rs = stmt.executeQuery()) {
                return rs.next();
            }
        }
    }

    /**
     * 插入监控日志
     */
    private boolean insertMonitorLog(Connection conn,
                                     EquipmentAlarmMonitorLogEnty entry,
                                     String deviceCode, Long productId,
                                     String productType) throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(insertLogSql)) {
            // 参数绑定
            stmt.setString(1, deviceCode);
            stmt.setString(2, entry.getAlertLevel());
            stmt.setString(3, entry.getAlertDimensions());
            stmt.setString(4, entry.getAlarmCover());
            stmt.setString(5, "connection_abnormal");
            stmt.setString(6, entry.getEventValue());
            stmt.setTimestamp(7, convertToSqlDate(entry.getAlarmTime()));
            stmt.setString(8, entry.getBlankingTool());
            stmt.setLong(9, productId);
            stmt.setString(10, productType);
            stmt.setString(11, entry.getMessageSuggestion());
            stmt.setString(12, entry.getAlertSource());
            stmt.setString(13, entry.getNetCode());
            stmt.setString(14, entry.getPersonnelHandler());
            stmt.setTimestamp(15, convertToSqlDate(entry.getHandlingTime()));
            stmt.setString(16, entry.getAlertDescription());
            stmt.setInt(17, 1); // 初始报警计数

            return stmt.executeUpdate() > 0;
        } catch (SQLException e) {
            logSqlError(e, "插入监控日志失败", deviceCode);
            return false;
        }
    }

    /**
     * 更新监控日志
     */
    private void updateMonitorLog(Connection conn,
                                  EquipmentAlarmMonitorLogEnty entry,
                                  String deviceCode) throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(updateLogSql)) {
            stmt.setInt(1, entry.getAlarmCount() + 1); // 报警计数+1
            stmt.setString(2, deviceCode);
            stmt.setString(3, "connection_abnormal");
            stmt.executeUpdate();
        } catch (SQLException e) {
            logSqlError(e, "更新监控日志失败", deviceCode);
        }
    }

    /**
     * 更新设备在线状态
     */
    private boolean updateOnlineStatus(Connection conn,
                                       EquipmentAlarmMonitorLogEnty entry,
                                       String deviceCode) throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(updateOnlineSql)) {
            stmt.setString(1, entry.getAlarmCover());
            stmt.setString(2, entry.getMessageSuggestion());
            stmt.setString(3, entry.getPersonnelHandler());
            stmt.setTimestamp(4, convertToSqlDate(entry.getHandlingTime()));
            stmt.setString(5, deviceCode);
            stmt.setString(6, "connection_abnormal");

            return stmt.executeUpdate() > 0;
        } catch (SQLException e) {
            logSqlError(e, "更新在线状态失败", deviceCode);
            return false;
        }
    }

    /**
     * 更新在线日志
     */
    private void updateOnlineLog(Connection conn,
                                 EquipmentAlarmMonitorLogEnty entry,
                                 String deviceCode) throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(updateLogSql)) {
            stmt.setInt(1, entry.getAlarmCount() + 1);
            stmt.setString(2, deviceCode);
            stmt.setString(3, "connection_abnormal");
            stmt.executeUpdate();
        } catch (SQLException e) {
            logSqlError(e, "更新在线日志失败", deviceCode);
        }
    }

    /**
     * 统一SQL异常处理
     */
    private void handleSQLException(SQLException e, int retryCount, String deviceCode) {
        if (e.getErrorCode() == DEADLOCK_ERROR_CODE && retryCount < MAX_RETRIES) {
            log.warn("设备 {} 发生死锁，正在重试 (第{}次)...", deviceCode, retryCount + 1);
        } else {
            log.error("设备 {} 不可恢复的数据库错误", deviceCode, e);
            throw new RuntimeException("数据库操作失败", e);
        }
    }

    /**
     * 统一SQL错误日志记录
     */
    private void logSqlError(SQLException e, String message, String deviceCode) {
        log.error("[{}] 设备: {} 错误码: {}", message, deviceCode, e.getErrorCode(), e);
    }

    /**
     * 日期转换工具方法
     */
    private Timestamp convertToSqlDate(Date date) {
        return date != null ? new Timestamp(date.getTime()) : null;
    }

    /**
     * 关闭资源
     */
    @Override
    public void close() throws Exception {
        super.close();
        if (jedis != null) jedis.close();
    }
}
